package peerToPeerFacebook.proj;

import java.io.Serializable;
import java.lang.reflect.InvocationTargetException;
import java.util.HashMap;
import java.util.Map;
import java.util.Observable;
import java.util.Observer;

/** RPC Server class. This class is responsible for handling remote invocation of methods invocations should come from
 * RPCClient class. There should be instance of this class per node.
 */
public final class RPCServer extends RPCBase implements IRPCServer	
{
	/** Object that provides real implementation of methods that should be invoked. RPC Server will delegate all real
	 * work to this instance. However, this object should not be concerned of enforcing AtMostOnce semantics. This is 
	 * done by RPC Server.
	 */
	private DynamicInvoker invoker;

	/** Determinates what was last execution state seen by this node for given client. It is possible that this one is
	 * null and we may need to load it from the disk.
	 * Index into this array is clientID, while value is state for given client.
	 */
    protected Map<Integer, ExecutionState> prevExecutionStates = new HashMap<Integer, ExecutionState>();
		
	/** Class that defines properties associated with particular call. This class is serialized to the hard drive to
	 * store state and make sure that crashes do not affect it.
	 * */ 
	protected static class ExecutionState implements Serializable
	{
		// classes implementing Serializable need to have some serialVersionUID defined
		// serialVersionUID is not being used by program as such
		private static final long serialVersionUID = -496338856859464727L;
		
		/** Sequence ID of the call */
		public long sequenceID = -1;
		
		/** Status of current call. Null means not started. We do not store call pending status in the disk.
		 */
		public Status status = null;
		
		/** Result of the call. Not valid unless status is Finished */
		public Serializable result = null;
		
		/** Callback function is such exists. If this one is present it will be called instead of invoking the function.
		 * There is no need to store this function in persistent storage as in case of crash we will not try to
		 * re-execute it.
		 */
		public transient CallPending pendingCall = null;
	}
	
	/** Constructs RPC Server
	 * 
	 * @param serverNode Node where RPC server is running on.
	 * @param dispatchObject Object that implements methods that RPC Server invokes.
	 */
	public RPCServer(INode serverNode, Object dispatchObject)
	{
		super(serverNode);
		this.invoker = new DynamicInvoker(dispatchObject);
	}

	/** Process packets and makes necessary changes in RPC server internal state.
	 */
	@Override
	protected void processPacket(int clientID, RPCPacketBase packet)
	{
		// See if this is even invoke message. If it is not we should just move on
		if (packet.action != RPCPacketBase.Action.Invoke)
			return;
		
		// Now that we know that this is invoke packet
		RPCInvokePacket invokePacket = (RPCInvokePacket)packet;
		
		ExecutionState prevState = prevExecutionStates.get(clientID);
		
		// See if we need to load this variable from the disk. It is possible that if we are command back from the
		// crash we do not have local variable initialized, but we may have a version on the disk.
		if (prevState == null)
		{
		    prevState = readFromDisk(getClientStorageFileName(clientID));
		    
		    // see if we can restore our local array too
		    if (prevState != null)
		        prevExecutionStates.put(clientID, prevState);
		}
		
		assert prevState == null || prevState.status != null : 
		    "If we have old execution states it's status should not be null";
		
		if (prevState != null)
			// See what sequence ID we have
			if (packet.sequenceID > prevState.sequenceID)
				makeNewCall(clientID, invokePacket);
			else if (packet.sequenceID == prevState.sequenceID)
			    // see if we need to re-send result or do pending call
			    if (prevState.pendingCall != null)
			        invokeLocally(clientID, prevState, invokePacket.method, invokePacket.params);
			    else
			        sendResult(clientID, prevState);
			else if (packet.sequenceID < prevState.sequenceID)
				node.logSynopticEvent("RPC:S: Received old packet form client(" + clientID + ") - dropping it." +
						"Current SeqID: " + prevState.sequenceID + ", packet: " + invokePacket);
			else
			    System.out.println("RPC-S: Dropping packet due too old sequence ID. Expecting at least " +
			            prevState.sequenceID + " got: " + packet.sequenceID);
		else
			makeNewCall(clientID, invokePacket);
	}

	protected String getClientStorageFileName(int clientID)
	{
		return "client-" + clientID;
	}
	
	/** Makes a new call and stores its state.
	 * 
	 * @param clientID Client making a call
	 * @param packet Packet describing the call
	 */
	protected void makeNewCall(int clientID, RPCInvokePacket packet)
	{
        // We already know that we need to invoke the call.
        // Regarding storage file it can be present or missing, in any case
        // we need to override what is in there with new state
        ExecutionState prevState = new ExecutionState();
        prevState.sequenceID = packet.sequenceID;
        
        // Initially we assume that function may or may not have been executed
        prevState.status = Status.AtMostOnce;

        // To guarantee AtMostOnce semantics in crash we need to flush our state before we invoke function. Any crash
        // before this point would imply that the function was not executed. However, this is not true as soon as we
        // try to call the function. This is the last place to write this information out.
        writeToDisk(getClientStorageFileName(clientID), prevState);
        prevExecutionStates.put(clientID, prevState);
        
        // now make the real call
        prevState = invokeLocally(clientID, prevState, packet.method, packet.params);        
	}
	

	/** Invokes command locally and stores result in persistent storage.
	 * 
	 * @param clientID Client making the call
	 * @param method Method to invoke
	 * @param params Set of parameters to use in invoke
	 */
	protected ExecutionState invokeLocally(int clientID, ExecutionState state, String method, Serializable[] params) throws RPCBroken
	{
		try {
		    state = invoke(clientID, state, method, params);
		} catch (InvocationTargetException e) {
			// Even RuntimeException and Errors exceptions from target will generate this exception, so we specifically
			// filter them out and let them pass - this sort of simulates server crash.
			Throwable t = e.getCause();
			if (t instanceof RuntimeException)
				throw (RuntimeException)t;
			else if (t instanceof Error)
				throw (Error)t;
			
			// If it is regular exception, tell the caller that something bad happened. We also return it to the caller
			state.pendingCall = null;
			state.status = Status.Exception;
			state.result = t;
		} catch (Throwable e1) {
			// All other exceptions gets reported to client
		    state.pendingCall = null;
		    state.status = Status.Exception;
		    state.result = e1;			
		}
		
        // again, make sure we have persistent record of what we did before we did something that could crash.
        writeToDisk(getClientStorageFileName(clientID), state);
        
        // now we can safely return the result to client. Crashes at this point would only mean that we lose answer to
        // client. Which is fine as client would just think that message was lost and re transmit it. At this point
        // we had stored enough information to reply back to it with result.
        sendResult(clientID, state);
		
		return state;		
	}

	protected void sendResult(int clientID, ExecutionState exe)
	{
		// We do not need to care here about such things as packet re-transmit and stuff. Client is responsible for
		// pulling results out of us.
		assert exe.status != null: "Packet should never be sent with status equals to null";
		
		// There are two cases that we can end up here - regular and pending call. We do not have a state for that
		// so we need to deal with this here.
		RPCReturnPacket packet = null;
		if (exe.pendingCall != null)
		{
		    assert exe.status == Status.AtMostOnce: "Call can be pending only if we have AtMostOnce semantics for now";
            packet = new RPCReturnPacket(exe.sequenceID, Status.CallPending, null);
		}
		else
            packet = new RPCReturnPacket(exe.sequenceID, exe.status, exe.result);
		
		// deal with packet encoding
		RPCException error = null;
		byte bits[] = null;
		try {
			bits = packet.encode();
		} catch (RPCError e) {
			error = e;
		}
		
		// If we are fine, check if we are within the size, so that we could send this packet
		if (bits.length > node.maxPacketSize() && error == null)
			error = new RPCPacketTooBig();
		
		// Now if we have error we need to re-encode packet
		if (error != null)
		{
			// Convert result to error now
			exe.status = Status.Exception;
			exe.result = error;
			
			// flush to disk to be consistent with any further calls.
			writeToDisk(getClientStorageFileName(clientID), exe);
			
			packet = new RPCReturnPacket(exe.sequenceID, exe.status, (Serializable)exe.result);
			try {
				bits = packet.encode();
			} catch (RPCError e) {
				// Ok now we have a problem. We cannot encode even a failure. There is nothing that RPC server can do
				// more. Time to bail out completely
				throw new RPCBroken("RPCPacketTooBig error is too big to send", e);
			}
		}
		
		node.send(clientID, RPCBase.Protocol.RPC.ID, bits);
	}
	
    public ExecutionState invoke(final int clientID, final ExecutionState state, final String methodName,
            final Serializable[] params) throws Throwable
    {
        assert state.status == Status.AtMostOnce: "This function can only be called if status has not been set in anyway";
        
        // see if we should use pending call approach
        if (state.pendingCall != null)
        {
            if (!state.pendingCall.completed())
                return state;
            
            // hey we can set the result. Let the client know that.
            state.result = state.pendingCall.result();
            state.status = Status.Success;
            state.pendingCall = null;
            return state;
        }
        
        // Ok now we know that this is not pending call, thus we invoke the function
        // Create new set of params that include Client ID
        Serializable[] fullParams = new Serializable[params.length + 1];
        fullParams[0] = clientID;
        for (int i = 0; i < params.length; i++)
            fullParams[i + 1] = params[i];
        
        // We may get either serializable or call pending, which is not serializable
        Object result = invoker.invoke(methodName, fullParams);
        
        // Check if this is pending call if so store it locally
        if (result instanceof CallPending)
        {
            state.pendingCall = (CallPending)result;
            state.pendingCall.addObserver(new Observer() {                
                @Override
                public void update(Observable o, Object arg)
                {
                    invokeLocally(clientID, state, methodName, params);
                }
            });
            
            // we do not change status. The reason is that if crash happens we want to make sure that we see null value
            // which means that call has not completed and thus this is AtMostOnce scenario.
            return state;
        }
        
        // Now we know that this is regular result. Thus we let the caller know that
        state.result = (Serializable)result;
        state.status = Status.Success;
        return state;
    }

}
